package main

import (
	"bufio"
	"context"
	"fmt"
	"log"
	"math"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	redisV8 "github.com/go-redis/redis/v8"
	"github.com/panjf2000/ants/v2"
	"github.com/rabbitmq/amqp091-go"
	yaml "gopkg.in/yaml.v3"
	_logger "sprixin.com/rxdb_checker/logger"
	model "sprixin.com/rxdb_checker/model"
)

func buildConfig() model.Configuration {
	content, err := os.ReadFile("config.yml")
	if err != nil {
		panic(err)
	}
	var conf model.Configuration
	if err := yaml.Unmarshal(content, &conf); err != nil {
		panic(err)
	}
	if err := conf.App.InitConfig(); err != nil {
		logger.Panicln(err)
	}
	return conf
}

func recreateDir(path string) error {
	// 检查目录是否存在
	_, err := os.Stat(path)
	if err == nil {
		// 目录存在，先删除
		fmt.Println("目录已存在，正在删除：", path)
		err = os.RemoveAll(path)
		if err != nil {
			return fmt.Errorf("删除目录失败: %w", err)
		}
	} else if !os.IsNotExist(err) {
		// 不是“文件不存在”错误，可能是其他错误
		return fmt.Errorf("检查目录失败: %w", err)
	}

	// 创建目录
	fmt.Println("正在创建目录：", path)
	err = os.MkdirAll(path, 0755)
	if err != nil {
		return fmt.Errorf("创建目录失败: %w", err)
	}

	return nil
}

func checkConfig(conf *model.Configuration) {
	curPath, err := os.Getwd()
	if err != nil {
		logger.Panicln("无法获取当前工作目录:", err)
	}
	if conf.App.MessageDir == nil {
		messageDir := filepath.Join(curPath, "messages")
		conf.App.MessageDir = &messageDir
	} else {
		*conf.App.MessageDir = filepath.Join(curPath, "messages")
	}

	// 检查消息目录是否存在,存在的话先删除再创建
	recreateDir(*conf.App.MessageDir)

	fmt.Printf("存储消息目录: %s\n", *conf.App.MessageDir)
}

func redisOperate2[T any, C any](redisClient *redisV8.Client, getVals func(*redisV8.Client) []T, redisCommand func(redisV8.Pipeliner, T) C, execResult func(int, C)) {

	pipline := redisClient.Pipeline()
	defer pipline.Close()

	cmds := make([]C, 0)
	for _, val := range getVals(redisClient) {
		// cmds = append(cmds, pipline.HMGet(ctx, "communicationlink&"+commId, "id", "hostgroupid"))
		cmds = append(cmds, redisCommand(pipline, val))
	}
	if _, err := pipline.Exec(ctx); err != nil {
		logger.Panicln(err)
	}

	for i, cmd := range cmds {
		execResult(i, cmd)
	}
}

func redisOperate(redisClient *redisV8.Client, setKey string, f func(redisV8.Pipeliner, string) *redisV8.SliceCmd, execResult func(*redisV8.SliceCmd)) {
	redisOperate2(redisClient, func(redisClient *redisV8.Client) []*interface{} {
		cmd := redisClient.SMembers(ctx, setKey)
		if cmd.Err() != nil {
			logger.Panicln(cmd.Err())
		}
		vals := cmd.Val()
		if vals == nil {
			logger.Panicln("无法从redis中获取到前置机组")
		}
		result := make([]*interface{}, len(vals))
		for i, v := range vals {
			var iface interface{} = v
			result[i] = &iface
		}
		return result
	}, func(p redisV8.Pipeliner, i *interface{}) *redisV8.SliceCmd {
		return f(p.Pipeline(), (*i).(string))
	}, func(i int, cmd *redisV8.SliceCmd) {
		if cmd.Err() != nil {
			logger.Panicln(cmd.Err())
		}
		if cmd.Val() == nil {
			logger.Panicln("前置机组id为空")
		}
		execResult(cmd)
	})
}

func redisOperate1(redisClient *redisV8.Client, setKey string, f func(redisV8.Pipeliner, string) *redisV8.SliceCmd, execResult func(*redisV8.SliceCmd)) {

	cmd := redisClient.SMembers(ctx, setKey)
	if cmd.Err() != nil {
		logger.Panicln(cmd.Err())
	}
	vals := cmd.Val()
	if vals == nil {
		logger.Panicln("无法从redis中获取到前置机组")
	}
	pipline := redisClient.Pipeline()
	defer pipline.Close()

	cmds := make([]*redisV8.SliceCmd, 0)
	for _, id := range vals {
		// cmds = append(cmds, pipline.HMGet(ctx, "communicationlink&"+commId, "id", "hostgroupid"))
		cmds = append(cmds, f(pipline, id))
	}
	if _, err := pipline.Exec(ctx); err != nil {
		logger.Panicln(err)
	}

	for _, cmd := range cmds {
		if cmd.Err() != nil {
			logger.Panicln(cmd.Err())
		}
		if cmd.Val() == nil {
			logger.Panicln("前置机组id为空")
		}

		execResult(cmd)
	}
}

func buildMapping() (map[int]*model.Communicationlink, map[int]*model.Communicationlink) {
	logger.Println("开始构建映射关系")
	// 通道
	cli := conf.Redis.GetClient()

	comm := make([]string, 0)
	redisOperate(cli, "communicationlink&all", func(p redisV8.Pipeliner, s string) *redisV8.SliceCmd {
		return p.HMGet(ctx, "communicationlink&"+s, "id", "hostgroupid")
	}, func(cmd *redisV8.SliceCmd) {
		vals := cmd.Val()
		sid := vals[0].(string)
		hostgroupid := vals[1].(string)
		hostGroupIds := strings.Split(*conf.App.HostGroupId, ",")
		if hostgroupid != "" && slices.Contains(hostGroupIds, hostgroupid) {
			comm = append(comm, fmt.Sprintf("%s-%s", sid, hostgroupid))
		}
	})
	logger.Printf("通道构建完成,通道数量：%d", len(comm))

	commMapping := make(map[int]*model.Communicationlink)
	for _, cidAndHostId := range comm {
		// if id, err := strconv.Atoi(com); err == nil {
		// 	commMapping[id] = &model.Communicationlink{Id: id}
		// } else {
		// 	logger.Panicln(err)
		// }
		ids := strings.Split(cidAndHostId, "-")
		cid, _ := strconv.Atoi(ids[0])
		hid, _ := strconv.Atoi(ids[1])
		commMapping[cid] = &model.Communicationlink{Id: cid, HostGroupId: hid}
	}

	// 获取路由与对应关系
	rtuMapping := make(map[int]*model.Communicationlink)
	redisOperate(cli, "ruroute&all", func(p redisV8.Pipeliner, s string) *redisV8.SliceCmd {
		return p.HMGet(ctx, "ruroute&"+s, "commlinkid", "rtuid")
	}, func(cmd *redisV8.SliceCmd) {
		vals := cmd.Val()
		commlinkid, _ := strconv.Atoi(vals[0].(string))
		rtuid, _ := strconv.Atoi(vals[1].(string))
		if com := commMapping[commlinkid]; com != nil {
			com.RtuId = rtuid
			rtuMapping[rtuid] = com
		}
	})
	logger.Println("路由构建完成")

	// analoginput
	redisOperate(cli, "analoginput&all", func(p redisV8.Pipeliner, s string) *redisV8.SliceCmd {
		if s == "0" {
			logger.Panicf(" 遥测点位id为0")
		}
		return p.HMGet(ctx, "analoginput&"+s, "rtuid", "pointnum", "equipmentcontainer_tableid", "equipmentcontainer_id", "type")
	}, func(cmd *redisV8.SliceCmd) {
		vals := cmd.Val()
		if hMgetFieldNotBeNull(vals, cmd.Args()) {
			return
		}
		id, _ := strconv.Atoi(strings.Split(cmd.Args()[1].(string), "&")[1])
		rtuid, _ := strconv.Atoi(vals[0].(string))
		pointnum, _ := strconv.Atoi(vals[1].(string))
		equTid, _ := strconv.Atoi(vals[2].(string))
		equId, _ := strconv.Atoi(vals[3].(string))
		meaId, _ := strconv.Atoi(vals[4].(string))
		if pointnum < 0 {
			return
		}
		if id == 0 {
			logger.Panicf("Rtuid:%d , pointnum: %d, 遥测点位id为0", rtuid, pointnum)
		}
		initCounter := int64(0)
		if com := rtuMapping[rtuid]; com != nil {
			com.AnaPoints = append(com.AnaPoints, &model.Point{Communicationlink: com, Id: id, PointNum: pointnum, Counter: &initCounter, IndexParttern: &model.ANA_POINT_INDEX, EquTid: equTid, EquId: equId, MeaId: meaId})
		}
	})
	logger.Println("遥测构建完成")

	// statusinput
	redisOperate(cli, "statusinput&all", func(p redisV8.Pipeliner, s string) *redisV8.SliceCmd {
		return p.HMGet(ctx, "statusinput&"+s, "rtuid", "pointnum", "equipmentcontainer_tableid", "equipmentcontainer_id", "type")
	}, func(cmd *redisV8.SliceCmd) {
		vals := cmd.Val()
		if hMgetFieldNotBeNull(vals, cmd.Args()) {
			return
		}
		id, _ := strconv.Atoi(strings.Split(cmd.Args()[1].(string), "&")[1])
		rtuid, _ := strconv.Atoi(vals[0].(string))
		pointnum, _ := strconv.Atoi(vals[1].(string))
		equTid, _ := strconv.Atoi(vals[2].(string))
		equId, _ := strconv.Atoi(vals[3].(string))
		meaId, _ := strconv.Atoi(vals[4].(string))
		if pointnum < 0 {
			return
		}
		if id == 0 {
			logger.Panicf("Rtuid:%d , pointnum: %d, 遥信点位id为0", rtuid, pointnum)
		}
		initCounter := int64(0)
		if com := rtuMapping[rtuid]; com != nil {
			com.StaPoints = append(com.StaPoints, &model.Point{Communicationlink: com, Id: id, PointNum: pointnum, Counter: &initCounter, IndexParttern: &model.STA_POINT_INDEX, EquTid: equTid, EquId: equId, MeaId: meaId})
		}
	})
	logger.Println("遥信构建完成")

	// accmulatorinput
	redisOperate(cli, "accmulatorinput&all", func(p redisV8.Pipeliner, s string) *redisV8.SliceCmd {
		return p.HMGet(ctx, "accmulatorinput&"+s, "rtuid", "pointnum", "equipmentcontainer_tableid", "equipmentcontainer_id", "type")
	}, func(cmd *redisV8.SliceCmd) {
		vals := cmd.Val()
		if hMgetFieldNotBeNull(vals, cmd.Args()) {
			return
		}
		id, _ := strconv.Atoi(strings.Split(cmd.Args()[1].(string), "&")[1])
		rtuid, _ := strconv.Atoi(vals[0].(string))
		pointnum, _ := strconv.Atoi(vals[1].(string))
		equTid, _ := strconv.Atoi(vals[2].(string))
		equId, _ := strconv.Atoi(vals[3].(string))
		meaId, _ := strconv.Atoi(vals[4].(string))
		if pointnum < 0 {
			return
		}
		if id == 0 {
			logger.Panicf("Rtuid:%d , pointnum: %d, 遥脉点位id为0", rtuid, pointnum)
		}
		initCounter := int64(0)
		if com := rtuMapping[rtuid]; com != nil {
			com.AccPoints = append(com.AccPoints, &model.Point{Communicationlink: com, Id: id, PointNum: pointnum, Counter: &initCounter, IndexParttern: &model.ACC_POINT_INDEX, EquTid: equTid, EquId: equId, MeaId: meaId})
		}
	})
	logger.Println("遥脉构建完成")

	return commMapping, rtuMapping
}

func hMgetFieldNotBeNull(vals []interface{}, keys []interface{}) bool {
	hasNil := false
	for i, v := range vals {
		if v == nil {
			hasNil = true
			logger.Printf("获取redis值存在nil情况,key=%s, field=%s \n", keys[1], keys[i+2])
		}
	}
	if hasNil {
		// 抛出异常
		// panic("获取redis值存在nil情况")

	}
	return hasNil
}

func buildPackageData(rtuid int, flag string, points []*model.Point) *string {
	var builder strings.Builder
	builder.WriteString(`{"`)
	builder.WriteString(flag)
	builder.WriteString(`":[`)
	timestamp := time.Now().Format("2006-01-02 15:04:05.000")
	for _, p := range points {
		(*p.Counter)++
		// timestamp := time.Now().Format("2006-01-02 15:04:05.000")
		for timestamp <= p.Timestamp {
			logger.Printf("通道: %d, 点位id: %d, 时间戳回拨,新：%s, 旧：%s, 重新获取...\n", p.Communicationlink.Id, p.Id, timestamp, p.Timestamp)
			timestamp = time.Now().Format("2006-01-02 15:04:05.000")
		}
		p.Timestamp = timestamp
		builder.WriteString(fmt.Sprintf(`[%d,%d,%d,"%s"],`, rtuid, p.PointNum, *p.Counter, timestamp))
	}

	// log.Println("写入消息数：", len(points))

	half := builder.String()
	result := half[:len(half)-1] + "]}"
	return &result
}

func doPublish(points []*model.Point, commlink *model.Communicationlink, packageSize int, flag string, routingKey string) int64 {
	length := len(points)
	if length == 0 {
		// logger.Printf("通道：%d, 表：%s, 没有需要写入的数据\n", commlink.Id, flag)
		return 0
	}
	// routingKey := fmt.Sprintf("g[%d].c[%d].analog.data", commlink.HostGroupId, commlink.Id)
	msgs := make([]*string, 0)
	var count int

	// ch := conf.Rabbitmq.GetChannel()
	// defer ch.Close()
	// confirms := ch.NotifyPublish(make(chan amqp091.Confirmation, 100))

	for i := 0; i < length; i += packageSize {
		min := int(math.Min(float64(length), float64(i+packageSize)))
		messageBody := points[i:min]
		message, _ := conf.App.MqMessage.GenerateMqMessage(commlink.RtuId, flag, messageBody)
		err := commlink.RabbitmqChannel.PublishWithContext(context.TODO(), "fepdata", routingKey, false, false,
			amqp091.Publishing{Body: []byte(*message)})
		if err != nil {
			log.Fatalln("发布消息失败", err, message)
		}
		msgs = append(msgs, message)
	}

	// 7. 等待确认
	// total := len(msgs)
	// ackCount := 0
	// timeout := time.After(10 * time.Second)
	// for ackCount < total {
	// 	select {
	// 	case confirm := <-confirms:
	// 		if confirm.Ack {
	// 			ackCount++
	// 		} else {
	// 			log.Fatalf("NACK: deliveryTag=%d", confirm.DeliveryTag)
	// 		}
	// 	case <-timeout:
	// 		log.Fatalf("等待确认超时，只确认了 %d 条", ackCount)
	// 	}
	// }

	count += len(points)
	file, err := os.OpenFile(filepath.Join(*conf.App.MessageDir, routingKey), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0777)
	if err != nil {
		logger.Panicln(err)
	}
	defer file.Close()
	for _, msg := range msgs {
		if _, err := file.WriteString(fmt.Sprintln(*msg)); err != nil {
			logger.Panicln(err)
		}
	}
	return int64(count)
}
func doMain(wg *sync.WaitGroup, app *model.App, commlink *model.Communicationlink, counter *atomic.Int64) {
	defer wg.Done()
	counter.Add(doPublish(commlink.AnaPoints, commlink, *app.PackageSize, "analoginput", fmt.Sprintf("g[%d].c[%d].analog.data", commlink.HostGroupId, commlink.Id)))
	counter.Add(doPublish(commlink.StaPoints, commlink, *app.PackageSize, "statusinput", fmt.Sprintf("g[%d].c[%d].status.data", commlink.HostGroupId, commlink.Id)))
	counter.Add(doPublish(commlink.AccPoints, commlink, *app.PackageSize, "accmulatorinput", fmt.Sprintf("g[%d].c[%d].accmul.data", commlink.HostGroupId, commlink.Id)))
}

func doCheckInner(commlinkId int, redisCli *redisV8.Client, flag string, points []*model.Point, errNum *atomic.Int32, errPointChan chan *model.Point) {
	results := make(map[string]*model.Point, len(points))
	redisOperate2(redisCli, func(c *redisV8.Client) []*interface{} {
		result := make([]*interface{}, len(points))
		for i, p := range points {
			var iface interface{} = p
			result[i] = &iface
		}
		return result
	}, func(p redisV8.Pipeliner, i *interface{}) *redisV8.StringCmd {
		point := (*i).(*model.Point)
		// key := flag + "&" + strconv.Itoa(point.Id)
		cmd, key := conf.App.RedisCommand.GetRedisValue(p, point, flag)
		results[key] = point
		return cmd
	}, func(i int, cmd *redisV8.StringCmd) {
		if err := cmd.Err(); err != nil {
			logger.Panicf("通道: %d, 表: %s, 检查时获取数据失败: %v", commlinkId, flag, err)
		}
		redisVal := cmd.Val()
		if redisVal == "" {
			logger.Panicf("通道: %d, 检查时获取数据失败: %v", commlinkId, cmd.Err())
		}
		key := cmd.Args()[1].(string)
		// rightVal := results[key]
		// redisIntVal, err := strconv.ParseFloat(redisVal, 32)
		// if err != nil {
		// 	logger.Panicf("通道: %d, 点位id: %s 转换redis值失败: %v", commlinkId, key, err)
		// }
		// if int64(redisIntVal) != *rightVal {
		// 	logger.Printf("通道: %d, 点位id: %s 检查时实际数:[%d]与redis[%d]数不对应.\n", commlinkId, key, *rightVal, int64(redisIntVal))
		// }
		rightVal := results[key].Timestamp
		if redisVal != rightVal {
			logger.Printf("通道: %d, 点位id: %s 检查时时间戳不对应，应为：[%s], redis: [%s].\n", commlinkId, key, rightVal, redisVal)
			errNum.Add(1)
			errPointChan <- results[key]
		}
	})
}

func doCheck(commlink *model.Communicationlink, wg *sync.WaitGroup, errNum *atomic.Int32, errPointChan chan *model.Point) {
	defer wg.Done()
	doCheckInner(commlink.Id, commlink.RedisCli, "analoginput", commlink.AnaPoints, errNum, errPointChan)
	doCheckInner(commlink.Id, commlink.RedisCli, "statusinput", commlink.StaPoints, errNum, errPointChan)
	doCheckInner(commlink.Id, commlink.RedisCli, "accmulatorinput", commlink.AccPoints, errNum, errPointChan)
}

func delFile(routingKey string) {
	if _, err := os.Stat(routingKey); err == nil {
		if err := os.Remove(routingKey); err != nil {
			logger.Panicln(err)
		}
	}
}

func filteAndBuildCommlinkMapping(commlinkMapping map[int]*model.Communicationlink) ([]*model.Communicationlink, int) {
	for _, commlink := range commlinkMapping {
		commlink.RabbitmqChannel = conf.Rabbitmq.GetChannel()
		commlink.RedisCli = conf.Redis.GetClient()
		anaLen := len(commlink.AnaPoints)
		staLen := len(commlink.StaPoints)
		accLen := len(commlink.AccPoints)
		anaMutationLen := int(float32(anaLen) * *conf.App.MutationRate)
		staMutationLen := int(float32(staLen) * *conf.App.MutationRate)
		accMutationLen := int(float32(accLen) * *conf.App.MutationRate)
		commlink.AnaPoints = commlink.AnaPoints[0:anaMutationLen]
		commlink.StaPoints = commlink.StaPoints[0:staMutationLen]
		commlink.AccPoints = commlink.StaPoints[0:accMutationLen]

		logger.Printf("通道：%d, 总点数:突变点数: 遥测[%d:%d], 遥信[%d:%d], 遥脉[%d:%d]\n", commlink.Id, anaLen, anaMutationLen, staLen, staMutationLen, accLen, accMutationLen)
		// 检查id0
		for _, p := range commlink.AnaPoints {
			if p.Id == 0 {
				logger.Panicf("通道: %d, 遥测点位id为0", commlink.Id)
			}
		}
		for _, p := range commlink.StaPoints {
			if p.Id == 0 {
				logger.Panicf("通道: %d, 遥测点位id为0", commlink.Id)
			}
		}
		for _, p := range commlink.AccPoints {
			if p.Id == 0 {
				logger.Panicf("通道: %d, 遥脉点位id为0", commlink.Id)
			}
		}

		// 如果文件存在
		delFile(filepath.Join(*conf.App.MessageDir, fmt.Sprintf("g[%d].c[%d].analog.data", commlink.HostGroupId, commlink.Id)))
		delFile(filepath.Join(*conf.App.MessageDir, fmt.Sprintf("g[%d].c[%d].status.data", commlink.HostGroupId, commlink.Id)))
		delFile(filepath.Join(*conf.App.MessageDir, fmt.Sprintf("g[%d].c[%d].accmul.data", commlink.HostGroupId, commlink.Id)))
	}

	totalPointNum := 0
	filteredCommlinkMapping := make([]*model.Communicationlink, 0)
	for _, commlink := range commlinkMapping {
		if !(len(commlink.AnaPoints) == 0 && len(commlink.StaPoints) == 0) {
			filteredCommlinkMapping = append(filteredCommlinkMapping, commlink)
			totalPointNum = totalPointNum + len(commlink.AnaPoints) + len(commlink.StaPoints)
		}
	}

	return filteredCommlinkMapping, totalPointNum
}

func startCtrlC() {
	// 创建一个 channel 用于接收信号
	sigs := make(chan os.Signal, 1)
	// 通知信号 channel，当接收到中断或终止信号时触发
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
	// 创建一个退出 channel
	// 启动 goroutine 等待信号
	go func() {
		sig := <-sigs
		fmt.Println()
		fmt.Println("收到信号:", sig)
		processStatus.Store(int32(CTRLC))
	}()

	fmt.Println("程序运行中，按 Ctrl+C 退出...")
}

func process(commlinks []*model.Communicationlink) {
	pool, err := ants.NewPool(len(commlinks), func(opts *ants.Options) {
		opts.PreAlloc = true
		opts.MaxBlockingTasks = 1000
	})
	if err != nil {
		logger.Panicln(err)
	}
	processStatus.Store(int32(RUNNING))
	startCtrlC()

	defer pool.ReleaseTimeout(time.Second * 10)
	for i := 0; i < *conf.App.Cyclecount || *conf.App.Cyclecount == -1; i++ {
		startTime := time.Now().UnixMilli()
		var wg sync.WaitGroup
		monitor := model.NewMonitor().Start()
		var counter atomic.Int64
		for _, commlink := range commlinks {
			wg.Add(1)
			if err := pool.Submit(func() {
				doMain(&wg, &conf.App, commlink, &counter)
			}); err != nil {
				logger.Panicln(err)
			}
		}
		wg.Wait()
		monitor.Println(logger, "第[%d]轮写入完成, 共写入点位: %d", i+1, counter.Load())
		costTime := time.Duration(time.Now().UnixMilli()-startTime) * time.Millisecond
		waitTime := time.Duration(*conf.App.Interval) * time.Millisecond
		if diff := waitTime - costTime; diff > 0 {
			time.Sleep(diff)
		}

		if processStatus.Load() == int32(CTRLC) {
			return
		}
	}
	processStatus.Store(int32(CHECKING))
}

func waitForAllDataSent(count int) {
	// 等待所有数据发送完毕
	logger.Println("等待所有数据发送完毕.............................................................................")
	if processStatus.Load() == int32(CTRLC) {
		count = 1
	}

	for i := 0; i < count && processStatus.Load() != int32(CTRLC); i++ {
		time.Sleep(time.Duration(1) * time.Second)
		logger.Printf(".")
	}
	logger.Println("所有数据已发送完毕...............................................................................")
}

func checkPointByIndex(errPointChan chan *model.Point, errNum *atomic.Int32, secondCheckFlag *sync.WaitGroup) {
	defer close(errPointChan)
	defer secondCheckFlag.Done()
	errPointCache := make([]*model.Point, 0)
	for {
		point := <-errPointChan
		if point == nil {
			break
		}
		errPointCache = append(errPointCache, point)
	}
	if len(errPointCache) == 0 {
		return
	}

	logger.Println("开始检查索引正确性.............................................................................")

	// 检查索引是否正确，若不正确则收集索引对应的点位，然后再去检查对应点位的时间戳
	redisIdKeys := make([]string, 0)
	redisPoints := make([]*model.Point, 0)
	redisOperate2(errPointCache[0].Communicationlink.RedisCli, func(c *redisV8.Client) []*model.Point {
		return errPointCache
	}, func(p redisV8.Pipeliner, point *model.Point) *redisV8.StringCmd {
		return p.Get(ctx, fmt.Sprintf(*point.IndexParttern, point.Communicationlink.RtuId, point.PointNum))
	}, func(i int, cmd *redisV8.StringCmd) {
		p := errPointCache[i]
		if err := cmd.Err(); err != nil {
			logger.Panicf("通道: %d, 检查时获取数据失败: %v", p.Communicationlink.Id, err)
		}
		key := cmd.Args()[1].(string)
		redisVal := cmd.Val()
		if redisVal == "" {
			logger.Printf("通道: %d, 点位: [%d], 未查到索引: [%s]\n", p.Communicationlink.Id, p.Id, key)
		}
		redisId := strings.Split(redisVal, "&")[1]
		if strconv.Itoa(p.Id) != redisId {
			logger.Printf("通道：[%d], 点位: [%d], 索引: [%s] 实际对应点位id: [%s]\n", p.Communicationlink.Id, p.Id, key, redisId)
			redisIdKeys = append(redisIdKeys, redisVal)
			redisPoints = append(redisPoints, p)
		} else {
			logger.Printf("严重：通道：[%d], 点位: [%d], 索引: [%s] 实际对应点位id: [%s], 与预期索引一致，但发生时间戳未变情况\n", p.Communicationlink.Id, p.Id, key, redisId)
		}
	})

	// 检查索引对应的点的时间戳是否正确
	if len(redisIdKeys) == 0 {
		return
	}
	redisOperate2(errPointCache[0].Communicationlink.RedisCli, func(c *redisV8.Client) []string {
		return redisIdKeys
	}, func(p redisV8.Pipeliner, idKey string) *redisV8.StringCmd {
		return p.HGet(ctx, idKey, "timestamp")
	}, func(i int, cmd *redisV8.StringCmd) {
		p := redisPoints[i]
		if err := cmd.Err(); err != nil {
			logger.Panicf("通道: %d, 检查时获取数据失败: %v", p.Communicationlink.Id, err)
		}
		key := cmd.Args()[1]
		redisVal := cmd.Val()
		if redisVal != p.Timestamp {
			logger.Printf("通道: %d, 点位id: %s 检查时时间戳不对应，应为：[%s], redis: [%s].\n", p.Communicationlink.Id, key, p.Timestamp, redisVal)
		} else {
			errNum.Add(-1)
		}
	})
}

func sencondCheck(errPointChan chan *model.Point, errNum *atomic.Int32) *sync.WaitGroup {
	var secondCheckFlag sync.WaitGroup
	secondCheckFlag.Add(1)
	go checkPointByIndex(errPointChan, errNum, &secondCheckFlag)
	return &secondCheckFlag
}

func goCheck(commlinks []*model.Communicationlink, totalPointNum int) {
	var errNum atomic.Int32
	var errPointChan = make(chan *model.Point, 100)

	secondCheckFlag := sencondCheck(errPointChan, &errNum)

	var wg sync.WaitGroup
	for _, commlink := range commlinks {
		wg.Add(1)
		go doCheck(commlink, &wg, &errNum, errPointChan)
	}
	wg.Wait()

	totalErrNum := errNum.Load()
	errPointChan <- nil    // 结束信号
	secondCheckFlag.Wait() // 等待索引检查完成

	if errNum.Load() > 0 {
		logger.Printf("总点位数: [%d], 共计错误点数: [%d], 索引错误：[%d], 实际错误点数:[%d]!\n", totalPointNum, totalErrNum, totalErrNum-errNum.Load(), errNum.Load())
	} else {
		logger.Println("恭喜你, 通过检查!")
	}
}

func goCheckIoTDB(commlinks []*model.Communicationlink, totalPointNum int) {

	// 1. 先检查点位中是否包含相同路径，有的话进行提示
	var exsitFlag map[string][]*model.Point = make(map[string][]*model.Point, 0)
	for _, commlink := range commlinks {
		tmp := append(commlink.AnaPoints, append(commlink.StaPoints, commlink.AccPoints...)...)
		for _, point := range tmp {
			iotdbpath := point.GetIoTDBPath()
			exsitFlag[iotdbpath] = append(exsitFlag[iotdbpath], point)
		}
	}
	pointInfoFile, err := os.OpenFile(filepath.Join(*conf.App.MessageDir, "iotdb_point_info.data"), os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		log.Fatalf("无法打开或创建文件: %v", err)
	}
	pointInfoFile.WriteString(fmt.Sprintln("#iotdbPath, pointId, rtuId, pointNum, type"))

	samePathLogFile, err := os.OpenFile(filepath.Join(*conf.App.MessageDir, "iotdb_same_path.log"), os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		log.Fatalf("无法打开或创建文件: %v", err)
	}
	for iotdbpath, points := range exsitFlag {
		if len(points) > 1 {
			txt := fmt.Sprintf("[%s] 存在多个点使用同一路径：%v \n", iotdbpath, points)
			log.Print(txt)
			if _, err := samePathLogFile.WriteString(txt); err != nil {
				log.Fatalf("无法写入文件: %v", err)
			}
		} else {
			// # iotdbPath, pointId, rtuId, pointNum, type
			if _, err := pointInfoFile.WriteString(fmt.Sprintf("%s,%d,%d,%d,%s\n", iotdbpath, points[0].Id, points[0].Communicationlink.RtuId, points[0].PointNum, strings.Split(*points[0].IndexParttern, "&")[0])); err != nil {
				log.Fatalf("无法写入文件: %v", err)
			}
		}
	}

	pointInfoFile.Close()
	samePathLogFile.Close()
	// 调用本地进程
	curDir, _ := os.Getwd()

	javaCommand := filepath.Join(curDir, "jre", "bin", "java.exe")
	jarPath := filepath.Join(curDir, *conf.App.CheckJarName)
	cmd := exec.Command(
		javaCommand,
		"-Dfile.encoding=UTF-8",
		"-jar",
		jarPath,
		fmt.Sprintf("--msg-dir=%s", *conf.App.MessageDir),
		fmt.Sprintf("--host=%s", *conf.IoTDB.Host),
		fmt.Sprintf("--port=%d", *conf.IoTDB.Port),
		fmt.Sprintf("--user=%s", *conf.IoTDB.User),
		fmt.Sprintf("--passwd=%s", *conf.IoTDB.Password))
	// 获取完整命令
	logger.Println(cmd.String())
	// 获取标准输出
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		logger.Panicf("无法获取标准输出: %v", err)
	}
	// 获取标准输出
	if err := cmd.Start(); err != nil {
		logger.Panicf("执行命令失败: %v", err)
	}

	// 读取标准输出
	scanner := bufio.NewScanner(stdout)
	for scanner.Scan() {
		// 解决中文乱码
		logger.Println(scanner.Text())
	}

	// 等待命令执行完成
	err = cmd.Wait()
	if err != nil {
		logger.Fatal("Java 程序执行出错：", err)
	}
}

func check(filteredCommlinkMapping []*model.Communicationlink, totalPointNum int) {
	switch *conf.App.AppVersion {
	case model.VERSION_060, model.VERSION_100, model.VERSION_100_REDIS:
		goCheck(filteredCommlinkMapping, totalPointNum)
	case model.VERSION_IOTDB:
		goCheckIoTDB(filteredCommlinkMapping, totalPointNum)
	default:
		logger.Panicf("不支持的版本: %s, 请检查配置文件", *conf.App.AppVersion)
	}
}

var ctx context.Context = context.Background()
var logger *log.Logger = _logger.GetLogger()
var processStatus atomic.Int32 // -1: 关闭, 0: 初始 1：正在写入, 2: 正在检查
var conf *model.Configuration

const (
	RUNNING    int = -1
	INIT       int = 0
	PUBLISHING int = 1
	CHECKING   int = 2
	CTRLC      int = 3
)

func main() {

	var c model.Configuration = buildConfig()
	checkConfig(&c)
	conf = &c
	logger.Println(conf)

	// build mapping/redis/rabbitmq
	commlinkMapping, _ := buildMapping()
	filteredCommlinkMapping, totalPointNum := filteAndBuildCommlinkMapping(commlinkMapping)

	// send message
	process(filteredCommlinkMapping)

	// wait for all data to be sent
	waitForAllDataSent(*conf.App.WaitTime)

	// check
	check(filteredCommlinkMapping, totalPointNum)
}
